%%%====================================================================
%%% File    : pipe_key_fold.erl
%%% Author  :  <vjache>
%%% Description: TODO Describe.
%%%
%%% Created : 30.03.2009
%%%====================================================================

-module(pipe_key_fold).

%%
%% Include files
%%
-include("../include/pipe.hrl").
%%
%% Exported Functions
%%
-export([
		 new/2,
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 init_accum/2,
		 test/1
		]).

%%====================================================================
%% API Functions
%%====================================================================


new({_FoldFun,_Acc0,_RecFun}=Arg,NextPipe) ->
	pipe:pipe_from_module(?MODULE, Arg, NextPipe).

open({{_FoldFun,_Acc0,_RecFun}=Param,NextPipe}) -> %%PipeState
	PrcNum=erlang:system_info(schedulers),
	{_Processes=list_to_tuple([
				   spawn_link(?MODULE, init_accum, [Param,self()])
				  ||_<-lists:seq(1,PrcNum)]),pipe:open(NextPipe)}.

push({_Processes,_NextPipeState}=S,?EOF) ->
	S;
push({Processes,_NextPipeState}=S,Item) 
  when is_tuple(Item) andalso size(Item)>1 ->
    PrcNum=erlang:phash2(element(2,Item), size(Processes))+1,
	element(PrcNum,Processes)!Item,
	S.
flush(S) ->
	S.
close({Processes,NextPipeState}) ->
	[Pid!?EOF || Pid<-tuple_to_list(Processes)],
	NextPipeStateFin=loop_rcv_results(size(Processes),NextPipeState),
	pipe:close(NextPipeStateFin),
	ok.

loop_rcv_results(0,NextPipeState) ->
	NextPipeState;
loop_rcv_results(N,NextPipeState) ->
	Self=self(),
	receive
		{?MODULE,Self,Rec} when is_tuple(Rec) ->
			loop_rcv_results(N,pipe:push(NextPipeState,Rec));
		{?MODULE,Self,?EOF} ->
			loop_rcv_results(N-1,NextPipeState)
	end.

			

%%====================================================================
%% Test Functions
%%====================================================================

test(L) ->
	S=pipe:open(new({fun({_,K,N},Acc)->
							Acc+N
					end,
					0,
					fun(K,Acc)->
							{summ,K,Acc}
					end},pipe_console:new())),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).


%%====================================================================
%% Local Functions
%%====================================================================

init_accum({_FoldFun,_Acc0,RecFun}=Arg,RetPid) ->
	?EOF=loop_accum(Arg,Tid=ets:new(accums, [])),
	ets:foldl(fun({Key,Acc},_)->
							  RetPid!{?MODULE,RetPid,RecFun(Key,Acc)}
			  end, 0, Tid),
	RetPid!{?MODULE,RetPid,?EOF}.
loop_accum({FoldFun,Acc0,RecFun},Tid) ->
	receive
		?EOF ->
			?EOF;
		Record ->
			Key=element(2,Record),
			case ets:lookup(Tid, Key) of
				[] ->
					ets:insert(Tid, {Key,FoldFun(Record,Acc0)}),
					loop_accum({FoldFun,Acc0,RecFun},Tid);
				[{Key,Acc}] ->
					ets:insert(Tid, {Key,FoldFun(Record,Acc)}),
					loop_accum({FoldFun,Acc0,RecFun},Tid)
			end
	end.
% erlang:system_info(schedulers)
